#!/usr/bin/env python
#==============================================================================
#
#  $Id$
#
"""
    Script for testing the ODB module.
"""
#
#   Copyright (C) 2006 Michael A. Muller
#
#   This file is part of ODB.
#
#   ODB is free software: you can redistribute it and/or modify it under the 
#   terms of the GNU Lesser General Public License as published by the Free 
#   Software Foundation, either version 3 of the License, or (at your option) 
#   any later version.
#
#   ODB is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU Lesser General Public License for more details.
#
#   You should have received a copy of the GNU Lesser General Public License 
#   along with ODB.  If not, see <http://www.gnu.org/licenses/>.
#
#==============================================================================

import shutil, unittest
import fixpath
import odb
import os

class test_odb(unittest.TestCase):

    def setUp(self):
        self.env = odb.Env('odb_test_db', prefer = odb.FULL)
	self.btree = self.env.getBTreeDB('btree')
	self.btreeData = [
	    ('test/1', 'first'),
	    ('test/2', 'first'),
	    ('test/3', 'first'),
	]
	for key, val in self.btreeData:
	    self.btree[key] = val

    def tearDown(self):
	del self.btree
	del self.env
	shutil.rmtree('odb_test_db')

    def testForwardCursor(self):
	cur = self.btree.cursor()
	cur.setToFirst()
	self.assertEquals([(key, val) for key, val in cur], self.btreeData)

    def testReverseCursor(self):
	cur = self.btree.cursor()
	cur.setToLast()
	data = self.btreeData[:]
	data.reverse()
	self.assertEquals([(key, val) for key, val in cur.reverse()], data)

    def testFindReverse(self):
	cur = self.btree.cursor()
	cur.setToKey('test/4')
	data = self.btreeData[:]
	data.reverse()
	self.assertEquals([(key, val) for key, val in cur.reverse()], data)

    def testCursorBefore(self):
	"""
	    Make sure setToKey() points the cursor to the first non-matching
	    key that's greater than the key specified, and is after any keys
	    less than it.
	"""
	# add an object before the key we're searching for to make sure we
	# don't get it
	self.btree['a'] = 'aval'

	cur = self.btree.cursor()
	cur.setToKey('s')
	self.assertEquals([(key, val) for key, val in cur], self.btreeData)

    def testCursorAfter(self):
	"""
	    Make sure we get nothing if the cursor is set to a key after the
	    last key in the table.
	"""
	cur = self.btree.cursor()
	cur.setToKey('zzz_after')
	self.assertRaises(StopIteration, cur.next)

    def testMultipleOpens(self):
	dbs = []
	for x in range(1, 1000):
	    db = self.env.getHashDB('hash')
	    dbs.append(db)

    def testBackup(self):
	"""
	    Tests backing up a cursor to make sure that the last object we got
	    going forward is the next object we get going backwards.
	"""
	cur = self.btree.cursor()
	cur.setToFirst()
	self.assertEquals(cur.next(), cur.prev())

	cur.setToLast()
	self.assertEquals(cur.prev(), cur.next())
    
    def testCursorTransactionClose(self):
	cur = self.btree.cursor()
	cur.setToFirst()
	for key, val in cur:
	    pass

	# remove the following line to make this test hang.  This is because
	# a cursor created without a transaction creates its own transaction,
	# as do the add and delete operations that follow
	cur = None

	for x in range(1, 20):
	    self.btree[`x`] = x
	for x in range(1, 20):
	    del self.btree[`x`]

    def testTransactionalCursorAndModification(self):
	txn = self.btree.getTxn()
	cur = self.btree.cursor(txn)
	cur.setToFirst()
	for key, val in cur:
	    pass

	for x in range(1, 20):
	    self.btree.set(txn, `x`, x)
	for x in range(1, 20):
	    self.btree.delete(txn, `x`)

    def testCursorTransactionAbort(self):
	txn = self.btree.getTxn()
	cur = self.btree.cursor(txn)
	txn.abort()

    def testEmptyKeys(self):
        self.btree[''] = 'empty'
    
    def testRecovery(self):
        
        # force generation of a few logfiles
        self.env.checkpoint()
        for i in range(1, 5):
            self.btree[`i`] = i
        self.env.checkpoint()
        for i in range(5, 10):
            self.btree[`i`] = i
        
        # remove the database file
        del self.env, self.btree
        os.remove('odb_test_db/btree')
        
        # make sure that the file set is what we expect
        files = os.listdir('odb_test_db')
        files.sort()
        expectedFiles = ['.full', '.mem', 'lock', 'log.000000001',
                         'log.000000002', 'log.000000003']
        self.assertEquals(expectedFiles, files)

        self.env = odb.Env('odb_test_db', prefer = odb.FULL)
        self.btree = self.env.getBTreeDB('btree')
        numbers = [(`x`, x) for x in range(1, 10)]
        allData = self.btreeData + numbers
        allData.sort()
        for (k1, v1), (k2, v2) in zip(allData, self.btree.cursor()):
            self.assertEquals((k1, v1), (k2, v2))

unittest.main()
# vim:set sw=4:
